package com.alpaca.recommend.hbase.init;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.io.IOException;

/**
 * Created by wangj on 2015/1/28.
 */
@Service
public class HBaseTables {
    private final static Logger logger = LoggerFactory.getLogger(HBaseTables.class);

    /**
     * 表结构,关联文档
     * http://word.baidu.com/share.php?id=e0d0uigxvi&qq-pf-to=pcqq.group
     */
    private final String[] tables = {
            "u_tags , t",
            "tag_items , t",
            "items , i",
            "users , u",
            "u_items , i",
            "u_recommends , r",
            "tag_user , t"
    };

    public static Configuration configuration;

    static {
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        configuration.set("hbase.zookeeper.quorum", "121.40.66.213,121.40.127.216,121.41.73.63");
//        configuration.set("hbase.zookeeper.quorum", "10.5.117.109");
    }


    public void createTables() throws IOException {
        HBaseAdmin admin = new HBaseAdmin(configuration);

        for (String tableStr : tables) {
            String[] tableDefine = tableStr.split(",");
            String tableName = tableDefine[0].trim();

            boolean exist = admin.tableExists(tableName);

            if (exist) {
                logger.error("table name :" + tableName + " is exist ,drop table!");
                admin.disableTable(tableName);
                admin.deleteTable(tableName);
            }

            TableName htableName = TableName.valueOf(tableName);
            HTableDescriptor tableDesc = new HTableDescriptor(htableName);

            for (int i = 1; i < tableDefine.length; i++) {
                HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(tableDefine[i].trim());

                tableDesc.addFamily(hColumnDescriptor);
            }
            admin.createTable(tableDesc);
            logger.info("create table success,name is:" + tableName);
        }
        admin.close();
    }
}
